package org.bongo.open.cumulo.procengine.broker;

import org.bongo.open.cumulo.core.context.CumuloContext;
import org.bongo.open.cumulo.core.message.MessageUnit;
import org.bongo.open.cumulo.core.utils.LazyIndexedObjectPool;
import org.bongo.open.cumulo.core.utils.PoolableObjectFactory;
import org.bongo.open.cumulo.procengine.broker.exceptions.ProcRunnerException;
import org.bongo.open.cumulo.procengine.runner.ProcRunner;
import org.bongo.open.cumulo.procengine.runner.SimpleProcRunnerFactory;


public class ProcEngineBroker {
	
	private LazyIndexedObjectPool<ProcRunner> runnerPool;
	private CumuloContext context;
	
	public ProcEngineBroker() {
	}
	
	public void initialize(CumuloContext c) {
		context = c;
		
		PoolableObjectFactory<ProcRunner> factory = new SimpleProcRunnerFactory(context);
		runnerPool = new LazyIndexedObjectPool<ProcRunner>(factory);
	}

	public void messageSubmitCallback(MessageUnit<?> message, ProcEnginePartitioner p)  {
		int runnerPartitionIndex = p.getPartition(message);
		
		// TODO this needs to be a separate process
		ProcRunner availableRunner = runnerPool.obtainFromPool(runnerPartitionIndex);
		
		try {
			availableRunner.runOnContainer(message);
		} catch (ProcRunnerException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		}
			
	}

	public void shutdown(CumuloContext c) {
		runnerPool.evictAll();
	}
	
	
}
